package com.yanggu.flink.tableapi_sql.define_table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Schema, TableDescriptor}

/**
 * 使用TableDescriptor来定义Table
 */
object TableDescriptorForTable {

  def main(args: Array[String]): Unit = {
    //1. 创建流式执行环境
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    //2. 创建表的执行环境
    val settings = EnvironmentSettings
      .newInstance()
      //使用流的方式
      .inStreamingMode()
      .build()
    val tableEnvironment = StreamTableEnvironment.create(environment, settings)

    //使用TableDescriptor来定义Table
    //类似于之前的老版本的connect的api
    val descriptor = TableDescriptor
      //支持的connector的名称需要在flink官方文档中查找
      //https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/overview/
      .forConnector("filesystem")
      .schema(Schema.newBuilder()
        //定义表的字段名称和数据类型
        .column("id", DataTypes.STRING())
        .column("timestamp", DataTypes.BIGINT())
        .column("temperature", DataTypes.DOUBLE())
        .build()
      )
      //定义格式, csv表示通过,分割, 需要添加对应的依赖flink-csv
      .format("csv")
      //文件系统需要指定路径, 也就是path配置
      .option("path", getClass.getResource("/sensor.txt").getPath)
      .build()

    //注册表
    tableEnvironment.createTemporaryTable("sensor", descriptor)

    //执行SQL
    tableEnvironment.sqlQuery(
      """
        |SELECT
        |   *
        |FROM
        |   sensor
        |WHERE
        |   id = 'sensor_1'
        |""".stripMargin)
      .execute()
      .print()

    //执行结果
    //+----+--------------------------------+----------------------+--------------------------------+
    //| op |                             id |            timestamp |                    temperature |
    //+----+--------------------------------+----------------------+--------------------------------+
    //| +I |                       sensor_1 |           1547718208 |                           36.2 |
    //| +I |                       sensor_1 |           1547718199 |                           35.8 |
    //| +I |                       sensor_1 |           1547718213 |                           30.9 |
    //| +I |                       sensor_1 |           1547718206 |                           32.0 |
    //| +I |                       sensor_1 |           1547718210 |                           29.7 |
    //+----+--------------------------------+----------------------+--------------------------------+

  }

}
